/*
 *  Licensed to the Apache Software Foundation (ASF) under one or more
 *  contributor license agreements.  See the NOTICE file distributed with
 *  this work for additional information regarding copyright ownership.
 *  The ASF licenses this file to You under the Apache License, Version 2.0
 *  (the "License"); you may not use this file except in compliance with
 *  the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */
package com.yuqih.common.websocket.monitor;

import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ExecutorService;

import javax.websocket.CloseReason;
import javax.websocket.Endpoint;
import javax.websocket.EndpointConfig;
import javax.websocket.MessageHandler;
import javax.websocket.Session;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.yuqih.common.json.JacksonObjectMapper;
import com.yuqih.common.websocket.IWebSocketManager;
import com.yuqih.common.websocket.IWebSocketMessageHandler;
import com.yuqih.common.websocket.SessionContainer;
import com.yuqih.common.websocket.WebSocketResponse;

public class MonitorWebSocketEndpoint extends Endpoint {

	private Log logger = LogFactory.getLog("WebSocket");

	// 不同的websocket用不同的Manager
	private IWebSocketManager webSocketManager;
	private ExecutorService executorService;

	@Autowired(required = false)
	private List<IMonitorHandler> monitorHandlers;

	private final IWebSocketMessageHandler handler = new MonitorWebSocketMessageHandler();

	private final static String ACTIONS = "ACTIONS";

	private final Map<MonitorAction, MonitorCachedContent> monitorActions = new HashMap<MonitorAction, MonitorCachedContent>();

	private IMonitorHandler getMonitorHandler(MonitorAction monitorAction) {
		if (monitorHandlers == null) {
			throw new NullPointerException();
		}

		for (IMonitorHandler handler : monitorHandlers) {
			if (StringUtils.equals(handler.getAction(), monitorAction.getAction())) {
				return handler;
			}
		}
		throw new IllegalArgumentException("找不到对应的处理类" + monitorAction);
	}

	@Override
	public void onOpen(Session session, EndpointConfig endpointConfig) {
		// HttpSession httpSession = (HttpSession) endpointConfig.getUserProperties().get(HttpSession.class.getName());
		SessionContainer container = webSocketManager.initialize(session, handler);
		container.getUserProperties().put(ACTIONS, new HashMap<MonitorAction, MonitorCachedContent>());
		session.addMessageHandler(new EchoMessageHandler(container));
	}

	@Override
	public void onClose(Session session, CloseReason closeReason) {
		webSocketManager.remove(session);
	}

	/**
	 * 去除不用的监控任务
	 */
	@SuppressWarnings("unchecked")
	private void clearMonitorActions() {

		Set<SessionContainer> containerSet = webSocketManager.getContainerSet();
		Set<MonitorAction> set = new HashSet<MonitorAction>();

		for (SessionContainer container : containerSet) {
			Map<MonitorAction, MonitorCachedContent> actionMap = (Map<MonitorAction, MonitorCachedContent>) container
					.getUserProperties().get(ACTIONS);

			synchronized (actionMap) {
				Iterator<Entry<MonitorAction, MonitorCachedContent>> iterator = actionMap.entrySet().iterator();
				while (iterator.hasNext()) {
					Entry<MonitorAction, MonitorCachedContent> entry = iterator.next();
					if (System.currentTimeMillis() - entry.getValue().time > 1000 * 60 * 1) {
						iterator.remove();
					} else {
						set.add(entry.getKey());
					}
				}
			}

		}

		synchronized (monitorActions) {
			Iterator<Entry<MonitorAction, MonitorCachedContent>> iterator = monitorActions.entrySet().iterator();
			while (iterator.hasNext()) {
				Entry<MonitorAction, MonitorCachedContent> entry = iterator.next();
				if (!set.contains(entry.getKey())) {
					iterator.remove();
				}
			}

		}

	}

	// quartz定时任务
	public void process() {

		clearMonitorActions();

		synchronized (monitorActions) {
			long time = System.currentTimeMillis();
			for (Entry<MonitorAction, MonitorCachedContent> entry : monitorActions.entrySet()) {
				MonitorAction action = entry.getKey();
				IMonitorHandler handler = getMonitorHandler(action);
				if (entry.getValue() != null && time - entry.getValue().time < handler.getInterval()) {
					continue;
				}

				executorProcess(action);

			}
		}

	}

	@SuppressWarnings("unchecked")
	private void process(MonitorAction action) {
		MonitorCachedContent content;
		Exception ex = null;
		synchronized (monitorActions) {
			content = monitorActions.get(action);
			if (content != null && Objects.equals(content.executing, Boolean.TRUE)) {
				return;
			}

			if (content == null) {
				content = new MonitorCachedContent();
				content.time = System.currentTimeMillis();
				content.executing = true;

				IMonitorHandler monitorHandler = getMonitorHandler(action);
				try {
					monitorHandler.preHandle(action);
				} catch (Exception e) {
					ex = e;
				}
				monitorActions.put(action, content);
			} else {
				content.time = System.currentTimeMillis();
				content.executing = true;
			}
		}

		Set<SessionContainer> set = webSocketManager.getContainerSet();
		if (ex != null) {
			handleException(action, ex, set);
			return;
		}

		// 执行任务
		Object body = null;
		try {
			IMonitorHandler monitorHandler = getMonitorHandler(action);
			body = monitorHandler.handle(action);
		} catch (RuntimeException e) {
			logger.warn("处理监控请求" + action.getUri() + "发生错误", e);
			ex = e;
		} finally {
			content.time = System.currentTimeMillis(); // 这样设置就意味着间隔时间了
			content.executing = false;
		}
		
		if (ex != null) {
			handleException(action, ex, set);
			return;
		} 

		// 将处理的值设到content
		content.body = body;
		// 异步发送消息，发到queue中

		for (SessionContainer container : set) {
			Map<MonitorAction, MonitorCachedContent> actionMap = (Map<MonitorAction, MonitorCachedContent>) container
					.getUserProperties().get(ACTIONS);
			synchronized (actionMap) {
				if (actionMap.containsKey(action)) {
					try {
						container.getQueue().put(new MonitorQueueContent(action, body));
					} catch (InterruptedException ignore) {
						// TODO 记日志
					}
				}
			}

		}

	}
	
	@SuppressWarnings("unchecked")
	protected void handleException(MonitorAction action, Exception ex, Set<SessionContainer> set) {

			for (SessionContainer container : set) {
				Map<MonitorAction, MonitorCachedContent> actionMap = (Map<MonitorAction, MonitorCachedContent>) container
						.getUserProperties().get(ACTIONS);

				String uri = null;
				synchronized (actionMap) {
					for (Entry<MonitorAction, MonitorCachedContent> entry : actionMap.entrySet()) {
						if (Objects.equals(action, entry.getKey())) {
							uri = entry.getKey().getUri();
							break;
						}
					}
				}
				try {
					if (uri != null) {
						webSocketManager.sendMessage(container.getSession(),
								new MonitorResponse(WebSocketResponse.ERROR, uri, "错误信息：" + ex.getMessage()));
					}
				} catch (IOException ignore) {
					// TODO 记日志
				}

			}
	}

	private void executorProcess(MonitorAction action) {
		executorService.execute(new Runnable() {

			@Override
			public void run() {
				process(action);
			}
		});

	}

	/**
	 * @param webSocketManager
	 *            the webSocketManager to set
	 */
	public void setWebSocketManager(IWebSocketManager webSocketManager) {
		this.webSocketManager = webSocketManager;
	}

	/**
	 * @param executorService
	 *            the executorService to set
	 */
	public void setExecutorService(ExecutorService executorService) {
		this.executorService = executorService;
	}

	private class MonitorWebSocketMessageHandler implements IWebSocketMessageHandler {

		@SuppressWarnings("unchecked")
		@Override
		public void handle(Object message, SessionContainer container) throws Exception{
			MonitorQueueContent content = (MonitorQueueContent) message;
				// 比较, 发送数据

				Map<MonitorAction, MonitorCachedContent> actionMap = (Map<MonitorAction, MonitorCachedContent>) container
						.getUserProperties().get(ACTIONS);

				String uri = null;
				MonitorCachedContent cachedContent = null;
				for (Entry<MonitorAction, MonitorCachedContent> entry : actionMap.entrySet()) {
					if (Objects.equals(entry.getKey(), content.getAction())) {
						uri = entry.getKey().getUri();
						cachedContent = entry.getValue();
						break;
					}
				}

				if (uri == null) {
					return;
				}

				IMonitorHandler monitorHandler = getMonitorHandler(content.getAction());
				Object result = monitorHandler.compare(content.getBody(), cachedContent.body);
				if (result == null) {
					result = Collections.EMPTY_LIST;
				} else if (!result.getClass().isArray() && !(result instanceof Collection)) {
					result = Arrays.asList(result);
				}

				webSocketManager.sendMessage(container.getSession(),
						new MonitorResponse(WebSocketResponse.OK, uri, result));
				cachedContent.body = content.getBody();
		}

	}

	private class EchoMessageHandler implements MessageHandler.Whole<String> {

		private final SessionContainer container;

		private EchoMessageHandler(SessionContainer container) {
			this.container = container;
		}

		@Override
		public void onMessage(String message) {
			logger.info(message);
			try {
				handleMessage(message);
			} catch (Exception e) {
				// TODO 记录日志
			}
		}

		@SuppressWarnings("unchecked")
		private void handleMessage(String message) throws Exception {
			// System.out.println("接收到消息:" + message);
			MonitorRequest request = JacksonObjectMapper.Default_Object_Mapper.readValue(message, MonitorRequest.class);
			MonitorAction action = new MonitorAction(request.getUri());

			Map<MonitorAction, MonitorCachedContent> actionMap = (Map<MonitorAction, MonitorCachedContent>) container
					.getUserProperties().get(ACTIONS);

			synchronized (actionMap) {
				Integer state = request.getState();
				if (Objects.equals(state, MonitorRequest.TYPE_REMOVE)) {
					actionMap.remove(action);
				} else if (Objects.equals(state, MonitorRequest.TYPE_ALIVE)) {
					// 如果里面有内容，修改时间即可
					MonitorCachedContent content = actionMap.get(action);
					if (content != null) {
						content.time = System.currentTimeMillis();
					} else {
						content = new MonitorCachedContent();
						content.time = System.currentTimeMillis();
						actionMap.put(action, content);
						executorProcess(action);
					}
				} else if (Objects.equals(state, MonitorRequest.TYPE_RESET)) {
					MonitorCachedContent content = actionMap.get(action);
					if (content != null) {
						content.time = System.currentTimeMillis();
						content.body = null;
						executorProcess(action);
					} else {
						content = new MonitorCachedContent();
						content.time = System.currentTimeMillis();
						actionMap.put(action, content);
						executorProcess(action);
					}
				} else if (Objects.equals(state, MonitorRequest.TYPE_HANDLE)) {
					// 如果里面有内容，修改时间即可
					MonitorCachedContent content = actionMap.get(action);
					if (content != null) {
						content.time = System.currentTimeMillis();
						executorProcess(action);
					} else {
						content = new MonitorCachedContent();
						content.time = System.currentTimeMillis();
						actionMap.put(action, content);
						executorProcess(action);
					}
				} else {
					throw new IllegalArgumentException("未知state");
				}
			}

		}
	}

}
